package ltd.hyangyang.rpc.core.remoting.server;


import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import ltd.hyangyang.rpc.common.bean.Request;
import ltd.hyangyang.rpc.common.constant.RemoteMsgConstant;
import ltd.hyangyang.rpc.core.serialize.RpcDecoder;
import ltd.hyangyang.rpc.core.serialize.RpcEncoder;

import java.util.concurrent.ConcurrentHashMap;
import java.util.logging.Logger;


public class RpcServerNetty {

    EventLoopGroup workGroup = new NioEventLoopGroup();
    EventLoopGroup bossGroup = new NioEventLoopGroup();
    //保存所有的channel
    public static ConcurrentHashMap<String,Channel> connectChannels ;

    static ConcurrentHashMap<String,Integer> freeRound = new ConcurrentHashMap<>();

    static Integer releaseRound = 10;

    Logger logger = Logger.getGlobal();

    public RpcServerNetty() throws InterruptedException {
        ServerBootstrap bootstrap = new ServerBootstrap();
        ChannelFuture channelFuture = bootstrap.group(workGroup,bossGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        pipeline.addLast(new RpcDecoder(Request.class));
                        pipeline.addLast(new RpcEncoder());
                        pipeline.addLast(new IdleStateHandler(100,100,3));

                        pipeline.addLast(new RpcServerHandler());
                    }
                })
                .bind(10000);
        Channel channel = channelFuture.channel().closeFuture().channel();

        connectChannels = RpcServerHandler.connectChannels;
    }

    public void sendPingMsg(Channel channel){
        Request ping = Request.builder()
                .msgType(RemoteMsgConstant.PING).build();

        //获取
        channel.writeAndFlush(ping).addListener((future)->{
            Integer round = freeRound.get(channel.id().asShortText());
            if (round==null){
               round=0;
            }
            round=round+1;
            freeRound.put(channel.id().asShortText(),round);
            if (!future.isSuccess()){
                //关闭channel
                logger.info("客户端关闭");
                connectChannels.remove(channel.id().asShortText());
                channel.close();
            }
            if (round>releaseRound){
                logger.info("channel长时间无通信，资源释放");
                channel.close();
            }
        });
    }

}
